package com.allen.flink.batch;

import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.configuration.Configuration;

/**
 * 功能:参数传递
 *
 * @date: 2020-03-24 15:00
 * @author: Allen
 * @version: 0.0.4-snapshot
 * @Email: allenZyhang@163.com
 * @since: JDK 1.8
 **/
public class Parameter {
    public static void main(String[] args) throws Exception {
        //获取运行环境
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        
        //读取数据
        DataSet<Integer> data = env.fromElements(1, 2, 3, 4, 5);
        
        // 构造方法传参
        //DataSet<Integer> result = data.filter(new MyFilter(3))
        
        //Configuration传参
        Configuration config = new Configuration();
        config.setInteger("limit", 3);
       
        /*DataSet<Integer> result = data.filter(new RichFilterFunction<Integer>() {
            private int limit;
            
            @Override
            public boolean filter(final Integer value) throws Exception {
                return value > limit;
            }
            
            @Override
            public void open(final Configuration config) throws Exception {
                limit = config.getInteger("limit", 0);
            }
        }).withParameters(config);
       */
        
        //全局传参
        env.getConfig().setGlobalJobParameters(config);
        
        DataSet<Integer> result = data.filter(new RichFilterFunction<Integer>() {
            private int limit;
            
            @Override
            public boolean filter(final Integer value) throws Exception {
                return value > limit;
            }
            
            @Override
            public void open(final Configuration parameters) throws Exception {
                super.open(parameters);
                ExecutionConfig.GlobalJobParameters global = getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
                Configuration config = (Configuration) global;
                limit = config.getInteger("limit", 0);
            }
        });
        
        result.print();
        
    }
    
    public static class MyFilter implements FilterFunction<Integer> {
        
        private int limit = 0;
        
        public MyFilter() {
        
        }
        
        public MyFilter(int limit) {
            this.limit = limit;
        }
        
        @Override
        public boolean filter(final Integer value) throws Exception {
            
            return value > limit;
        }
    }
}
